package demo.mapreduce.scala

object WordCount {
  def main(args: Array[String]): Unit = {
    val content:List[String] = List("qiusuo zhao hello spark", "zhao spark scala qiusuo zhao hello");

    var statisticResult:List[(String,Int)] = null;

    statisticResult = statisticByMap(content)(sortASC);
    printList(statisticResult);
    println("// -----")
    statisticResult = statisticByReduce(content)(sortDESC);
    printList(statisticResult);
  }


  /**
   * WordCount， 统计给出的文本合集中 按照制定规则匹配的文本序列 出现的次数.
   * @param content
   * @param split
   * @param sortCount
   * @return
   */
  def statisticByMap(content : List[String],split:() => String = defaultDelimiter)
                        (implicit sortCount:Int => Int = sortASC): List[(String,Int)] = {
    var statistic : List[(String,Int)] =
      content
        // 将每个元素按照空格打散！输出 List[String]
        .flatMap((item :String) => item.split(split.apply()))
        // 将打散后的元素映射为元祖，元祖对每个单词每次出现都标记为1 List[(String,Int)]
        .map((item:String) => (item,1))
        // 按照元祖第一个元素进行分组，将相同的单词归为一组, Map[String,List[(String,Int)]]
        .groupBy((item:(String,Int)) => item._1)
        // 对分组后的结果再映射，求每组中元祖的个数。Map[String,(String,Int)]
        .map((item:(String,List[(String,Int)])) => (item._1,item._2.size))
        // 将vales打成List[(String,Int)].
        .toList
        // 排序
        .sortBy((item:(String,Int)) => sortCount(item._2))
    return statistic;
  }


  def statisticByReduce(content:List[String],split:() => String = defaultDelimiter)
                       (implicit sortCount:Int => Int = sortASC):List[(String,Int)] = {
    val statistic:List[(String,Int)] = content
      .flatMap(item => item.split(split.apply()))
      .map(item => (item, 1))
      .groupBy(item => item._1)
      .map(item => (item._1, item._2.reduce((left, right) => (left._1, left._2 + right._2))))
      .values
      .toList
      .sortBy(item => sortCount.apply(item._2))
    return statistic;
  }


  //排序函数
  val sortASC:Int => Int = item => item;
  val sortDESC:Int => Int = sortASC_FUN _;

  def sortASC_FUN(sortItem :Int):Int = {
    return -sortItem;
  }

  //默认分隔符函数
  val defaultDelimiter: () => String = () => " ";


  //输出.
  def printList[Item](list:List[Item]):Unit = {
    if(list != null && list.size > 0){
      val iterator: Iterator[Item] =  list.iterator;
      while (iterator.hasNext)
        println(iterator.next())
    }
  }
}
